package cn.zwx.business.goods.mq.consumer;

import cn.hutool.core.collection.CollectionUtil;
import cn.zwx.base.enums.MqTopicEnum;
import cn.zwx.business.goods.service.GoodsService;
import cn.zwx.order.req.DeductionRepertoryReq;
import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import static cn.zwx.business.goods.constants.CommonConstant.APPLICATION_NAME;
import static cn.zwx.business.goods.constants.CommonConstant.NAME_SERVER;

/**
 * @description: 创建订单消费者
 * @projectName:rocket-mq-example
 * @see:cn.zwx.business.goods.mq.consumer
 * @author:zhangwenxue
 * @createTime:2020/10/15 14:04
 * @version:1.0
 */
@Component
public class OrderConsumer {

    private static final Logger logger =  LoggerFactory.getLogger(OrderConsumer.class);

    @Autowired
    private GoodsService goodsService;

    private DefaultMQPushConsumer consumer;

    /**
     * @description 消费组 每个消费组一条消息只能被消费一次
     **/
    private static final String consumerGroup = "goods_consumer_group";

    /**
     * @description 消费端处理逻辑
     * @author zhangwenxue
     * @createTime 2020/10/15 14:16
     **/
    public OrderConsumer() throws MQClientException {
        //设置消费组
        consumer = new DefaultMQPushConsumer(APPLICATION_NAME);
        // 添加服务器地址
        consumer.setNamesrvAddr(NAME_SERVER);
        /**
         * 1. CONSUME_FROM_LAST_OFFSET：第一次启动从队列最后位置消费，后续再启动接着上次消费的进度开始消费
         2. CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
         3. CONSUME_FROM_TIMESTAMP：第一次启动从指定时间点位置消费，后续再启动接着上次消费的进度开始消费
         以上所说的第一次启动是指从来没有消费过的消费者，如果该消费者消费过，那么会在broker端记录该消费者的消费位置，
         如果该消费者挂了再启动，那么自动从上次消费的进度开始
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         * CLUSTERING：默认模式，同一个ConsumerGroup(groupName相同)每个consumer只消费所订阅消息的一部分内容，同一个ConsumerGroup里所有的Consumer消息加起来才是所
         *  订阅topic整体，从而达到负载均衡的目的
         * BROADCASTING：同一个ConsumerGroup每个consumer都消费到所订阅topic所有消息，也就是一个消费会被多次分发，被多个consumer消费。
         *
         */
        consumer.setMessageModel(MessageModel.BROADCASTING);
        // 添加订阅号
        consumer.subscribe(MqTopicEnum.ORDER_TOPIC.getTopic(), "*");
        // 监听消息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            //不等于空才处理业务逻辑
            if (CollectionUtil.isNotEmpty(msgs)){
                MessageExt msg = msgs.get(0);
                String message = new String(msg.getBody());
                DeductionRepertoryReq repertoryReq = JSONObject.parseObject(message, DeductionRepertoryReq.class);
                logger.info("消费端消费消息，请求参数:{}",repertoryReq);
                try {
                    //扣减库存逻辑处理
                    goodsService.deductionRepertory(repertoryReq);
//                    int i = 10 / 0;
                    logger.info("扣减库存支成功!!!!");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    logger.info("消费失败，进行重试，重试到一定次数 那么将该条记录记录到数据库中，进行如果处理");
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

        });

        consumer.start();
       logger.info("consumer start ...");
    }



}
